/*
 *  Queen Mary University of London MOM System
 *  
 *  Copyright Queen Mary University of London
 *  
 *  Authors:
 *  Bob Chew (bob.chew@elec.qmul.ac.uk)
 *  Beatriz Viñal Murciano (beavimu@gmail.com)
 *  John Bigham (john.bigham@elec.qmul.ac.uk)
 *  
 *  
 *  Portions of this software uses Apache Qpid
 *    http://qpid.apache.org
 *  
 *  Apache Qpid license:
 * --------------------------------------------------------------
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 * -------------------------------------------------------------
 * 
 */
package users;

import javax.jms.JMSException;
import javax.jms.TopicSubscriber;

import org.apache.qpid.client.AMQConnection;
import org.apache.qpid.client.AMQSession;
import org.apache.qpid.client.AMQTopic;
import org.apache.qpid.client.message.JMSTextMessage;

/**
 * Subscriber that gets a message from a queue in a mirror broker when the
 * subscriber it belongs to receives a message in the queue with the same
 * name in the primary broker.
 */
public class MirrorSubscriber {

    /** Subscription to subscribe to. */
    private Subscription subscription;
    /**
     * Getter for the variable 'subscription'.
     * @return variable 'subscription'
     */
    public Subscription getSubscription() {
        return subscription;
    }
    /**
     * Setter for the variable 'subscription'.
     * @param subscription  new value of the variable 'subscription'
     */
    public void setSubscription(final Subscription subscription) {
        this.subscription = subscription;
    }

    /** Connection to the mirror broker. */
    private AMQConnection connection;
    /** Session in the connection to the mirror broker. */
    private AMQSession session;
    /**
     * Getter for the variable 'connection'.
     * @return variable 'connection'
     */
    public AMQConnection getConnection() {
        return connection;
    }
    /**
     * Setter for the variable 'connection'.
     * @param connection  new value of the variable 'connection'
     */
    public void setConnection(final AMQConnection connection) {
        this.connection = connection;
    }
    /**
     * Getter for the variable 'session'.
     * @return variable 'session'
     */
    public AMQSession getSession() {
        return session;
    }
    /**
     * Setter for the variable 'session'.
     * @param session  new value of the variable 'session'
     */
    public void setSession(final AMQSession session) {
        this.session = session;
    }

    /** Subscriber to the topic of the subscription. */
    private TopicSubscriber subscriber;
    /**
     * Getter for the variable 'subscriber'.
     * @return variable 'subscriber'
     */
    public TopicSubscriber getSubscriber() {
        return subscriber;
    }
    /**
     * Setter for the variable 'subscriber'.
     * @param subscriber  new value of the variable 'subscriber'
     */
    public void setSubscriber(final TopicSubscriber subscriber) {
        this.subscriber = subscriber;
    }

    /** Number of messages to read. */
    private int numMsgs = 0;
    /**
     * Threshold timestamp of the received messages.
     * The mirror subscriber must never get a message with a timestamp
     * higher that this one.
     */
    private long thTimestamp = 0;


    /**
     * Constructor.
     *
     * @param subscriber  who controls the mirror reader
     * @param connection  to use
     * @param subscription  to subscribe to
     */
    public MirrorSubscriber(final Subscriber subscriber,
            final AMQConnection connection, final Subscription subscription) {

        this.connection = connection;
        this.subscription = subscription;

        //subscribe to subscription
        try {
            //create session in the connection
            setSession((AMQSession) getConnection().createSession(
                    true, AMQSession.AUTO_ACKNOWLEDGE));
            //create the topic URL
            String topicURL = "topic://" + subscriber.getMOM_EXCHANGE()
                              + "/?routingkey='" + subscription.getTopic()
                              + "'&exclusive='false'&autodelete='true'";
            //create the topic
            AMQTopic topic = (AMQTopic) getSession().createTopic(topicURL);
            //create a subscriber
            setSubscriber(getSession().createDurableSubscriber(topic,
                    subscription.getId(),
                    subscription.getMsgSelector(), false));
            //commit the changes
            getSession().commit();

            System.out.println("Mirror of " + subscriber.getClientId()
                    + " subscribed to topic "
                    + getSubscriber().getTopic().getTopicName()
                    + " in broker " + getHost() + " at port " + getPort());

            //create the thread that receives messages
            ReceiveThread rt = new ReceiveThread();
            //start it
            rt.start();
        } catch (JMSException e) {
            System.err.println("Error subscribing to " + getHost() + ": " + e);
        }

    }

    /**
     * Get the host the mirror subscriber is connected to.
     *
     * @return  IP address of the host the mirror subscriber is connected to
     */
    public String getHost() {
        return getConnection().getActiveBrokerDetails().getHost();
    }

    /**
     * Get the port the mirror subscriber is connected to.
     *
     * @return  port the mirror subscriber is connected to
     */
    public int getPort() {
        return getConnection().getActiveBrokerDetails().getPort();
    }

    /**
     * Add 1 to the number of messages to be read.
     *
     * @param timestamp  timestamp of the message received by the subscriber
     */
    protected void addMsgToRead(final long timestamp) {
        thTimestamp = timestamp;
        numMsgs++;
    }

    /**
     * Remove the subscription in the mirror broker.
     * @throws JMSException  if there is any problem
     */
    protected void removeSubscription() throws JMSException {
        String subscriptionId = getSubscription().getId();
        
        //unsubscribe
        getSession().unsubscribe(subscriptionId);
        //commit the changes
        getSession().commit();
    }

    /**
     * Closes the subscriber of the mirror subscriber.
     *
     * @throws JMSException  if there is any problem
     */
    protected void close() throws JMSException {
        //close topic subscriber
        getSubscriber().close();
        //close session
        getSession().close();
    }

    /**
     * Thread to receive a message.
     */
    class ReceiveThread extends Thread {
        /** Constructor. */
        public ReceiveThread() { }

        //run
        @Override
        public void run() {

            while (!getSession().isClosed() && !getSession().isClosing()) {
                if (numMsgs > 0) {

//                    try { //TODO
//                        sleep(3000);
//                    } catch (InterruptedException e1) {
//                        e1.printStackTrace();
//                    }

                    JMSTextMessage message = null;
                    try {

                        //subscriber receives message
                        message = (JMSTextMessage) getSubscriber().receive();

                        //if message is older than the one received by the
                        //subscriber
                        if (message.getJMSTimestamp() <= thTimestamp) {
                            //commit the changes (remove message from queue)
                            getSession().commit();
                            System.out.println("Mirror received message \""
                                    + message.getText() + "\" from "
                                    + message.getJMSDestination());
                        } else { //message newer than the one received by the
                                 //subscriber
                            //roll back the changes (keep message in queue)
                            getSession().rollback();
                            System.out.println("Mirror refused receiving "
                                    + "message \"" + message.getText() + "\" "
                                    + "from " + message.getJMSDestination());
                        }

                        //decrease number of messages to read
                        numMsgs--;

                    } catch (JMSException e) {
                        //error receiving message
                        System.err.println("Error receiving message from "
                                           + getHost() + ": " + e);
                        e.printStackTrace();
                    }
                }
            }

        }
    }

}
